package com.hadoop.mr.reduceJoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @program: hadoop
 * @description: tableMapper 用于获取文本中的数据进行mapper操作
 * @author: Qiang.Ye
 * @date: 2019-06-09 12:03
 **/
public class TableMapper extends Mapper<LongWritable,Text,Text,TableBean> {

    Text k = new Text();
    TableBean v = new TableBean();

    @Override
    protected void map(LongWritable key,
                       Text value,
                       Context context) throws IOException, InterruptedException {

        //通过contetxt找到InputSplit的子类FileSplit  注意:转类型
        FileSplit split = (FileSplit) context.getInputSplit();
        //找到文件名
        String name = split.getPath().getName();

        String line = value.toString();
        System.out.println("===1111===");
        //订单表
        if(name.startsWith("order")){
            System.out.println("===2222===");
            String[] fields = line.split("\t");
            v.setOrderId(fields[0]);
            v.setPid(fields[1]);
            v.setAmount(Integer.parseInt(fields[2]));
            v.setPname("");
            v.setFlag("0");
            k.set(fields[1]);
        }else {
            System.out.println("===3333===");
            //按照产品表处理
            String[] fields = line.split("\t");
            v.setPid(fields[0]);
            v.setPname(fields[1]);
            v.setAmount(0);
            v.setOrderId("");
            v.setFlag("1");
            k.set(fields[0]);
        }
        context.write(k,v);
    }
}
